/*-------------------------------------------------------------------------
 *
 * tcn.c
 *	  PostgreSQL 的触发器变更通知支持
 *
 * Portions Copyright (c) 2011-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  contrib/tcn/tcn.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/htup_details.h"
#include "commands/async.h"
#include "commands/trigger.h"
#include "executor/spi.h"
#include "lib/stringinfo.h"
#include "utils/rel.h"
#include "utils/syscache.h"

PG_MODULE_MAGIC;

/*
 * 从 s（源）复制到 r（结果），用 q（引号）字符包裹
 * 并且对找到的任何引号字符进行双重处理。
 */
static void fc_strcpy_quoted(StringInfo fc_r, const char *fc_s, const char fc_q)
{
	appendStringInfoCharMacro(fc_r, fc_q);
	while (*fc_s)
	{
		if (*fc_s == fc_q)
			appendStringInfoCharMacro(fc_r, fc_q);
		appendStringInfoCharMacro(fc_r, *fc_s);
		fc_s++;
	}
	appendStringInfoCharMacro(fc_r, fc_q);
}

/*
 * triggered_change_notification
 *
 * 此触发函数将发送数据修改的通知，包含主键值。
 * 通道将是 "tcn"，除非触发器是通过参数创建的，在这种情况下将使用该参数。
 */
PG_FUNCTION_INFO_V1(triggered_change_notification);

Datum triggered_change_notification(PG_FUNCTION_ARGS)
{
	TriggerData *fc_trigdata = (TriggerData *) fcinfo->context;
	Trigger    *fc_trigger;
	int			fc_nargs;
	HeapTuple	fc_trigtuple;
	Relation	fc_rel;
	TupleDesc	fc_tupdesc;
	char	   *fc_channel;
	char		fc_operation;
	StringInfo	fc_payload = makeStringInfo();
	bool		fc_foundPK;

	List	   *fc_indexoidlist;
	ListCell   *fc_indexoidscan;

	/* 确保作为触发器调用 */
	if (!CALLED_AS_TRIGGER(fcinfo))
		ereport(ERROR,
				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
				 errmsg("triggered_change_notification: must be called as trigger")));

	/* 并且在更改后调用 */
	if (!TRIGGER_FIRED_AFTER(fc_trigdata->tg_event))
		ereport(ERROR,
				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
				 errmsg("triggered_change_notification: must be called after the change")));

	/* 并且对每一行调用 */
	if (!TRIGGER_FIRED_FOR_ROW(fc_trigdata->tg_event))
		ereport(ERROR,
				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
				 errmsg("triggered_change_notification: must be called for each row")));

	if (TRIGGER_FIRED_BY_INSERT(fc_trigdata->tg_event))
		fc_operation = 'I';
	else if (TRIGGER_FIRED_BY_UPDATE(fc_trigdata->tg_event))
		fc_operation = 'U';
	else if (TRIGGER_FIRED_BY_DELETE(fc_trigdata->tg_event))
		fc_operation = 'D';
	else
	{
		elog(ERROR, "triggered_change_notification: trigger fired by unrecognized operation");
		fc_operation = 'X';		/* 消除编译器警告 */
	}

	fc_trigger = fc_trigdata->tg_trigger;
	fc_nargs = fc_trigger->tgnargs;
	if (fc_nargs > 1)
		ereport(ERROR,
				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
				 errmsg("triggered_change_notification: must not be called with more than one parameter")));

	if (fc_nargs == 0)
		fc_channel = "tcn";
	else
		fc_channel = fc_trigger->tgargs[0];

	/* 获取元组数据 */
	fc_trigtuple = fc_trigdata->tg_trigtuple;
	fc_rel = fc_trigdata->tg_relation;
	fc_tupdesc = fc_rel->rd_att;

	fc_foundPK = false;

	/*
	 * 从 relcache 获取表的索引 OID 列表，并在 pg_index 系统缓存中查找
	 * 每一个，直到找到一个标记为主键的索引
	 * （希望不会超过一个这样的索引）。
	 */
	fc_indexoidlist = RelationGetIndexList(fc_rel);

	foreach(fc_indexoidscan, fc_indexoidlist)
	{
		Oid			fc_indexoid = lfirst_oid(fc_indexoidscan);
		HeapTuple	fc_indexTuple;
		Form_pg_index fc_index;

		fc_indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(fc_indexoid));
		if (!HeapTupleIsValid(fc_indexTuple))	/* 不应该发生 */
			elog(ERROR, "cache lookup failed for index %u", fc_indexoid);
		fc_index = (Form_pg_index) GETSTRUCT(fc_indexTuple);
		/* 我们只关心它是否是主键并且有效 */
		if (fc_index->indisprimary && fc_index->indisvalid)
		{
			int			fc_indnkeyatts = fc_index->indnkeyatts;

			if (fc_indnkeyatts > 0)
			{
				int			fc_i;

				fc_foundPK = true;

				fc_strcpy_quoted(fc_payload, RelationGetRelationName(fc_rel), '"');
				appendStringInfoCharMacro(fc_payload, ',');
				appendStringInfoCharMacro(fc_payload, fc_operation);

				for (fc_i = 0; fc_i < fc_indnkeyatts; fc_i++)
				{
					int			fc_colno = fc_index->indkey.values[fc_i];
					Form_pg_attribute fc_attr = TupleDescAttr(fc_tupdesc, fc_colno - 1);

					appendStringInfoCharMacro(fc_payload, ',');
					fc_strcpy_quoted(fc_payload, NameStr(fc_attr->attname), '"');
					appendStringInfoCharMacro(fc_payload, '=');
					fc_strcpy_quoted(fc_payload, SPI_getvalue(fc_trigtuple, fc_tupdesc, fc_colno), '\'');
				}

				Async_Notify(fc_channel, fc_payload->data);
			}
			ReleaseSysCache(fc_indexTuple);
			break;
		}
		ReleaseSysCache(fc_indexTuple);
	}

	list_free(fc_indexoidlist);

	if (!fc_foundPK)
		ereport(ERROR,
				(errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),
				 errmsg("triggered_change_notification: must be called on a table with a primary key")));

	return PointerGetDatum(NULL);	/* 在触发器之后；值无关紧要 */
}
